package com.wzy.chapter05;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransformRichFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Bob", "./home", 3000L),
                new Event("Alice", "./prod?id=100", 4000L),
                new Event("Alice", ".//prod?id=200", 5000L),
                new Event("Alice", ".//prod?id=300", 6000L)
        );
        SingleOutputStreamOperator<Integer> result = stream.map(new MyRichFunction());
        result.print();
        env.execute();
    }
    public static class MyRichFunction extends RichMapFunction<Event,Integer>{
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            System.out.println("open生命周期被调用"+getRuntimeContext().getIndexOfThisSubtask()+"任务启动");
        }

        @Override
        public void close() throws Exception {
            super.close();
            System.out.println("close生命周期被调用"+getRuntimeContext().getIndexOfThisSubtask()+"任务结束");
        }

        @Override
        public Integer map(Event event) throws Exception {
            return event.url.length();
        }
    }
}
